package xuecheng.tool.mq.redis;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.core.annotation.AnnotationUtils;

import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import xuecheng.tool.mq.MqListener;
import xuecheng.tool.nosql.redis.CpxRedisAutoConfig;

/**
 * 自定义 RedisMq 配置
 *
 * @author PengXueCheng
 * @version 1.0.0
 **/
@SpringBootConfiguration
@AutoConfigureAfter(CpxRedisAutoConfig.class)
@EnableConfigurationProperties(RedisMqProperties.class)
// @ConditionalOnProperty(prefix = "cpx.redismq", value = "enable", havingValue = "true")
// @ConditionalOnBean(value = RedisUtils.class)
@Slf4j
public class CpxRedisMqBeanPostProcessor implements BeanPostProcessor {

    @Autowired
    private ConsumerFactory consumerFactory;

    /**
     * 注册消费者
     *
     * @param bean
     * @param beanName
     * @return
     * @throws BeansException
     */
    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        // 对所有 MqListener 的所有子类且添加有 RedisMQMessageListener 类注解的做 RedisMq 主题订阅
        Class<?> targetClass = AopUtils.getTargetClass(bean);
        RedisMqMessageListener listener = AnnotationUtils.findAnnotation(targetClass, RedisMqMessageListener.class);
        boolean assignableFrom = MqListener.class.isAssignableFrom(targetClass);
        boolean ok = listener != null && assignableFrom;
        if (ok) {
            register(listener, (MqListener)bean);
        }
        return bean;
    }

    /**
     * RedisMq 主题订阅
     *
     * @param listenerAnnotation
     *            队列类注解
     * @param mqListener
     *            队列消息监听接口
     */
    private void register(RedisMqMessageListener listenerAnnotation, MqListener mqListener) {
        String topic = listenerAnnotation.topic();
        int consumerCount = listenerAnnotation.consumerCount();

        Class messageClass = getMessageClass(mqListener);

        // 开启多个消费者实例
        for (int i = 0; i < consumerCount; i++) {
            subscribe(topic, messageClass, mqListener);
        }
        log.info("RedisMq Consumer Started. topic: {}, consumerCount: {}", topic, consumerCount);
    }

    /**
     * 获取 {@link MqListener} 子类的的 消息实体类
     *
     * @param redisMQListener
     *            {@link MqListener} 子类实例
     * @return 消息实体类
     */
    private Class getMessageClass(MqListener redisMQListener) {
        Class listenerClass = MqListener.class;
        Class<?> beanClass = redisMQListener.getClass();

        ParameterizedType parameterizedType = getParameterizedType(listenerClass, beanClass.getGenericInterfaces());
        Type[] actualTypeArguments = parameterizedType.getActualTypeArguments();
        Class messageClass;
        try {
            messageClass = (Class)actualTypeArguments[0];
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            throw new IllegalArgumentException(
                listenerClass.getName() + "实现类泛型类型参数列表必须只有一个Class参数.eg:MqListener<Object>", e);
        }
        return messageClass;
    }

    /**
     * 从实体类的继承的所有接口获取预期继承的接口类型的参数化类型
     *
     * @param targetClass
     *            预期继承的接口类型
     * @param genericInterfaces
     *            实体类的继承的所有接口
     * @return 预期继承的接口类型的参数化类型
     */
    private <T> ParameterizedType getParameterizedType(Class<T> targetClass, Type[] genericInterfaces) {
        List<Type> interfaceList = Arrays.stream(genericInterfaces).filter(type -> {
            if (ParameterizedType.class.isAssignableFrom(type.getClass())) {
                ParameterizedType parameterizedType = (ParameterizedType)type;
                return parameterizedType.getRawType() == targetClass;
            }
            return false;
        }).collect(Collectors.toList());
        if (interfaceList.size() == 1) {
            return (ParameterizedType)interfaceList.get(0);
        }
        return null;
    }

    /**
     * 消息订阅
     *
     * @param topic
     *            队列主题
     * @param messageClass
     *            消息实体字节码
     * @param listener
     *            监听器
     * @param <T>
     */
    private <T> void subscribe(@NonNull String topic, @NonNull Class<T> messageClass, @NonNull MqListener<T> listener) {
        consumerFactory.start(topic, messageClass, listener);
    }

}
